﻿using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
using Microsoft.VisualStudio.TestPlatform.UnitTestFramework;
using XcoAppSpaces.Core;
using XcoAppSpaces.Core.Ports;
using System.Threading;

namespace test.XcoAppSpaces.Core
{
	[TestClass]
	public class testPublishSubscribe
	{
		[TestMethod]
		public void TestPublishSubscribe1()
		{
			using (XcoAppSpace space1 = new XcoAppSpace("tcp.port=8000"),
				space2 = new XcoAppSpace("tcp.port=8001"))
			{
				space1.RunWorker<PNewsAgency, TransientNewsAgency>("worker");
				PNewsAgency na = space2.ConnectWorker<PNewsAgency>("localhost:8000/worker");

				Port<PNewsAgency.OnNewsArrived> sub0 = new Port<PNewsAgency.OnNewsArrived>();
				na.Post(new Subscribe<PNewsAgency.OnNewsArrived>(sub0));
				Task.Delay(200).Wait(); // wait for subscription to be processed before next publication

				na.Post(new PNewsAgency.PublishNews { PublisherName = "peter", NewsItem = "greetings from peter" });
				Task.Delay(200).Wait();

				Port<PNewsAgency.OnNewsArrived> sub1 = new Port<PNewsAgency.OnNewsArrived>();
				na.Post(new Subscribe<PNewsAgency.OnNewsArrived>(sub1));
				Task.Delay(200).Wait();

				na.Post(new PNewsAgency.PublishNews { PublisherName = "mary", NewsItem = "greetings from mary" });
				Task.Delay(200).Wait();

				na.Post(new PNewsAgency.PublishNews { PublisherName = "john", NewsItem = "greetings from john" });
				Task.Delay(200).Wait();

				SpinWait.SpinUntil (() => sub0.ItemCount == 3, 2000);
				Assert.AreEqual(3, sub0.ItemCount);
				SpinWait.SpinUntil (() => sub1.ItemCount == 2, 2000);
				Assert.AreEqual(2, sub1.ItemCount);
				
				PNewsAgency.OnNewsArrived news;
				news = sub0.Receive(1000).Result;
				Assert.AreEqual("peter", news.PublisherName);
				Assert.AreEqual("greetings from peter", news.NewsItem);
				news = sub0.Receive(1000).Result;
				Assert.AreEqual("mary", news.PublisherName);
				Assert.AreEqual("greetings from mary", news.NewsItem);
				news = sub0.Receive(1000).Result;
				Assert.AreEqual("john", news.PublisherName);
				Assert.AreEqual("greetings from john", news.NewsItem);

				news = sub1.Receive(1000).Result;
				Assert.AreEqual("mary", news.PublisherName);
				Assert.AreEqual("greetings from mary", news.NewsItem);
				news = sub1.Receive(1000).Result;
				Assert.AreEqual("john", news.PublisherName);
				Assert.AreEqual("greetings from john", news.NewsItem);
			}
		}

		public class PNewsAgency : PortSet<PNewsAgency.PublishNews, Subscribe<PNewsAgency.OnNewsArrived>, Unsubscribe<PNewsAgency.OnNewsArrived>>
		{
			//[Serializable]
			public class PublishNews
			{
				public string PublisherName;
				public string NewsItem;
			}

			//[Serializable]
			public class OnNewsArrived
			{
				public string PublisherName;
				public string NewsItem;

				private readonly DateTime arrivedAt = DateTime.Now;

				public DateTime ArrivedAt
				{
					get { return this.arrivedAt; }
				}
			}
		}

		public class TransientNewsAgency : PNewsAgency
		{
			private readonly XcoPublisher<OnNewsArrived> publisher = new XcoPublisher<OnNewsArrived>();

			[XcoConcurrent]
			public void ProcessPublishNews(PublishNews news)
			{
				OnNewsArrived signal = new OnNewsArrived { PublisherName = news.PublisherName, NewsItem = news.NewsItem };
				publisher.Publish(signal);
			}
		}
	}
}
